package com.shujia.rec.compute

import com.shujia.rec.common.Constants
import com.shujia.rec.funaction.ItemCfSimilarityFunaction
import com.shujia.rec.util.{KafkaUtil, LogUtil}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object ItemCfSimilarity {
  def main(args: Array[String]): Unit = {

    /**
      * 商品相似度计算
      *
      */


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //获取kafka source
    val kafkaSource = KafkaUtil.getKafaSoure("ItemCfHistory", Constants.KAFKA_TOPIC)

    //读所有数据
    kafkaSource.setStartFromLatest()

    val logDS = env.addSource(kafkaSource)

    //过滤脏数据
    val filterDS = logDS.filter(log => LogUtil.verifyLog(log))

    //将每一行数据转换成实体类
    val entryDS = filterDS.map(log => LogUtil.toEntry(log))


    //计算相关性
    entryDS.flatMap(new ItemCfSimilarityFunaction)


    env.execute("ItemCfSimilarity")

  }
}
